package com.learn.kafka;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.IntegerDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

/**
 * kafka 消费者
 *
 * @Author: cong zhi
 * @CreateDate: 2021/3/2 15:38
 * @UpdateUser: cong zhi
 * @UpdateDate: 2021/3/2 15:38
 * @UpdateRemark: 修改内容
 * @Version: 1.0
 */
public class KafkaConsumerDemo extends Thread {

    private KafkaConsumer<Integer, String> kafkaConsumer;

    private String topic;

    @Override
    public void run() {
        kafkaConsumer.subscribe(Collections.singleton(this.topic));
        while (true) {
            ConsumerRecords<Integer, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
            consumerRecords.forEach(record -> {
                System.out.println(record.key() + "->" + record.value() + "->" + record.offset());
            });

        }
    }

    public KafkaConsumerDemo(String topic) {
        Properties properties = new Properties();
        // kafka 集群地址
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.110.17:9092");
        properties.put(ConsumerConfig.CLIENT_ID_CONFIG, "KafkaConsumerDemo2");
        // 消费组
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "kafka-gid");
        // 关闭自动提交offset
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        // 设置间隔时间
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "10000");
        // 超时时间
        properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
        // 反序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // 这个属性，它能够消费昨天数据

        this.kafkaConsumer = new KafkaConsumer<Integer, String>(properties);
        this.topic = topic;

    }

    public static void main(String[] args) {
        new KafkaConsumerDemo("test-lisa").start();

//        System.out.println(("KafkaConsumerDemo2".hashCode())%50);
    }

}
